package com.sali.service.data_processor;

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.sali.commons.disruptor.*;
import com.sali.commons.exception.ProcessChainHandlerInitException;
import com.sali.entity.ProcessDefinition;
import com.sali.entity.TransferObject;
import com.sali.service.data_processor.chain.ProcessChainContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * @description: 数据处理初始化操作
 * @author: sali
 * @date: 2025/3/2 下午1:45.
 * @version: 1.0
 */
public class DataProcessContext {

    public static Logger log = LoggerFactory.getLogger(DataProcessContext.class);

    // 处理规则的名称
    private String name;

    // 处理流程配置信息
    private ProcessDefinition processConfig;

    private Disruptor<DataProcessEvent> disruptor;

    private RingBuffer<DataProcessEvent> ringBuffer;

    public DataProcessContext(ProcessDefinition processConfig) {
        this.processConfig = processConfig;
        init(processConfig);
        this.ringBuffer = this.disruptor.getRingBuffer();
    }

    //  根据预置的配置信息，初始化数据处理流程中的相关组件：处理模块、数据队列、处理线程
    private void init(ProcessDefinition processConfig) {
        try {
            // 处理化处理流程的名称, 默认名称：default
            this.name = getTaskName();

            // 初始化处理链路
            ProcessChainContext chainContext = new ProcessChainContext();
            chainContext.loadProcessChain(processConfig.getPath(), processConfig);

            // 初始化disruptor
            Integer queueLen = to2N(processConfig.getQueueLen());
            this.disruptor = new Disruptor<>(
                    new DataProcessEventFactory(),
                    queueLen,
                    new ProcessThreadFactory(this.name),
                    ProducerType.MULTI,
                    new SleepingWaitStrategy()
            );
            Integer threadNum = processConfig.getThreadNum();
            DataProcessHandler[] dphs = new DataProcessHandler[threadNum];
            for (int i = 0; i < threadNum; i++) {
                dphs[i] = new DataProcessHandler(chainContext);
            }
            this.disruptor.handleEventsWithWorkerPool(dphs);
            this.disruptor.setDefaultExceptionHandler(new DataProcessExceptionHandler());
            this.disruptor.start();

        } catch (ProcessChainHandlerInitException e) {
            log.error("[处理链路：{}] 初始化失败,请检查配置信息:{}", name, processConfig.toString());
            throw new RuntimeException(e);
        }
    }

    public void put(TransferObject obj) {
        long seq = this.ringBuffer.next();
        try {
            DataProcessEvent dataProcessEvent = this.ringBuffer.get(seq);
            dataProcessEvent.setVal(obj);
        } catch (Exception e) {
            // 处理异常
            log.error("[数据处理流程：{}]提交数据异常", name, e);
        } finally {
            this.ringBuffer.publish(seq);
        }
    }


    private String getTaskName() {
        String taskName = processConfig.getName();
        if (taskName != null && !"".equals(taskName.trim())) {
            return taskName;
        }
        List<String> protocolType = processConfig.getProtocolType();
        if (protocolType == null || protocolType.isEmpty()) {
            //优化硬编码
            return "default";
        }
        String join = String.join("-", protocolType);
        return join.substring(0, Math.min(10, join.length()));
    }

    // 将一个数值转为刚好小于等于当前值的2^N
    private static Integer to2N(Integer queueLen) {
        // 如果queueLen小于等于0，直接返回1
        if (queueLen <= 0) {
            return 1;
        }

        int power = 1;
        while (power <= queueLen) {
            // 左移一位相当于乘以2
            power <<= 1;
        }

        return power >> 1;
    }

}
